package disruptor;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.dsl.Disruptor;
import disruptor.event.ValueEvent;
import disruptor.event.handler.ValueEventHandler;
import disruptor.factory.ValueEventFactory;

import java.util.concurrent.Executors;

/**
 * @ClassName SingleProducerMultipleConsumerWorkPool
 * @PackageName disruptor
 * @projectName myThread
 * @Description 工作池模式示例：单生产者、多消费者（每个事件仅被一个消费者处理）
 * @Author games
 * @Date 2025/9/8 下午5:22
 * @Version 1.0
 */
public class SingleProducerMultipleConsumerWorkPool {
    public static void run() throws InterruptedException {
        // 创建 Disruptor - 单生产者模式
        Disruptor<ValueEvent> disruptor = new Disruptor<>(
                new ValueEventFactory(),
                1024,
                Executors.defaultThreadFactory(),
                com.lmax.disruptor.dsl.ProducerType.SINGLE,
                new com.lmax.disruptor.SleepingWaitStrategy()
        );

        // 创建工作处理器（WorkHandler）
        WorkHandler<ValueEvent>[] workHandlers = new WorkHandler[]{
                new ValueEventHandler("Worker-1"),
                new ValueEventHandler("Worker-2"),
                new ValueEventHandler("Worker-3")
        };

        // 使用工作池模式连接事件处理器
        disruptor.handleEventsWithWorkerPool(workHandlers);

        // 启动 Disruptor
        disruptor.start();

        // 获取 Ring Buffer
        RingBuffer<ValueEvent> ringBuffer = disruptor.getRingBuffer();

        // 生产事件
        for (int i = 0; i < 10; i++) {
            long sequence = ringBuffer.next();
            try {
                ValueEvent event = ringBuffer.get(sequence);
                event.setValue(i);
            } finally {
                ringBuffer.publish(sequence);
            }
        }

        Thread.sleep(2000); // 等待所有事件处理完成
        disruptor.shutdown();
    }
}
